package com.zhaosc.rx.eventbus;

import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

public class RxBackPressureEventBus {

    private FlowableProcessor<Object> bus;

    private RxBackPressureEventBus() {
        //把非线程安全的PublishSubject包装成线程安全的SerializedSubject
        bus = PublishProcessor.create().toSerialized();
    }

    public static RxBackPressureEventBus getDefault() {
        return SingletonHolder.INSTANCE;
    }

    private static class SingletonHolder {
        public static volatile RxBackPressureEventBus INSTANCE = new RxBackPressureEventBus();
    }

    /**
     * 发送事件
     *
     * @param event 事件对象
     */
    public void post(Object event) {
        if (bus.hasSubscribers()) {
            bus.onNext(event);
        }
    }

    /**
     * 监听事件
     *
     * @return 特定类型的Observable
     */
    public Flowable<Object> observe() {
        return bus;
    }

    /**
     * 监听事件
     *
     * @param event 事件对象
     * @param <T>       事件类型
     * @return 特定类型的Observable
     */
    public <T> Flowable<T> observe(Class<T> event) {
        return bus.ofType(event);
    }
}